﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Psm.PubSub.StateCache;
using Psm.PubSub.Serialization;
using Psm.PubSub.Serialization.Prefixed;
using System.Collections;
using Psm.Util;

namespace Psm.PubSub.Messaging
{
    public class DataMessage : PubSubMessage
    {   
        private string[] _data;
        private ByteBuffer _payload;
        private string _key;
        public string Topic { get { return Schema == null ? null : Schema.Topic; } }
        public Schema Schema { get; private set; }
        public BitArray DataFields { get; private set; }

        public override PubSubMessageType MessageType
        {
            get { return PubSubMessageType.DATA_MESSAGE; }
        }

        public string Key
        {
            get
            {
                if (_key != null) return _key;
                else
                {
                    StringBuilder keyBuilder = new StringBuilder();
                    for (int index = 0; index < Schema.KeyFields.Count; index++)
                        if (keyBuilder.Length > 1)
                            keyBuilder.Append("_").Append(_data[Schema.KeyFields[index].Index] ?? "NULL");
                        else
                            keyBuilder.Append(_data[Schema.KeyFields[index].Index] ?? "NULL");
                    _key = keyBuilder.ToString();
                    return _key;
                }
            }

            set
            {
                _key = value;
            }
        }

        internal DataMessage(ByteBuffer payload)
        {   
            _payload = payload;
            StringMessageDeserializer serializer = new StringMessageDeserializer(payload);
            string topic = serializer.GetField();
            Schema = SchemaCollection.AllSchemas[topic];
            _data = new string[Schema.Fields.Count];
            this.DataFields = new BitArray(Schema.Fields.Count);
            this.Key = serializer.GetField();
            while (serializer.HasMore())
            {
                int fieldIndex;
                string value;
                serializer.GetTuple(out fieldIndex, out value);
                MarkDataField(fieldIndex);
                _data[fieldIndex] = value;
            }
        }

        public DataMessage(Schema schema)
        {
            this.Schema = schema;
            _data = new string[schema.Fields.Count];
            this.DataFields = new BitArray(Schema.Fields.Count);
        }
        
        public override bool Serialize(ByteBuffer buffer)
        {
            if (_payload != null) { return buffer.Write(_payload); }
            StringMessageSerializer serializer = new StringMessageSerializer(PubSubMessageType.DATA_MESSAGE, buffer);
            if( !serializer.AppendField(Topic) ) return false;
            if( !serializer.AppendField(Key) ) return false;
            for (int index = 0; index < this.Schema.Fields.Count; index++)
            {
                if (DataFields[index])
                {   
                    if( !serializer.AppendTuple(index, _data[index]) ) return false;
                }
            }
            return true;
        }

        public void AddData(string name, string value)
        {
            SchemaField field = this.Schema.GetField(name);
            if (field != null)
            {
                MarkDataField(field.Index);
                _data[field.Index] = value;
            }
        }

        public string GetData(string name)
        {
            string value = null;
            SchemaField field = this.Schema.GetField(name);
            if (field != null)
                value =  _data[field.Index];
            return value;
        }

        public string GetData(SchemaField field)
        {
            string value = null;
            if (field != null)
                value = _data[field.Index];
            return value;
        }

        public void MergeMessage(DataMessage other)
        {
            for (int index = 0; index < Schema.Fields.Count; index++)
            {
                if (other.DataFields.Get(index))
                {
                    MarkDataField(index);
                    _data[index] = other._data[index];
                }
            }
        }

        private void MarkDataField(int index)
        {
            if (_data[index] == null)
                this.DataFields.Set(index, true);
        }

        internal void ClearPayload()
        {
            _payload = null;
        }
    }
}
